package localmq

import (
	"errors"
	"fmt"
	"os"
	"os/signal"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"hundsun.com/hsl/hschain/common/types"
	prototypes "hundsun.com/hsl/hschain/protos/types"
	drivers "hundsun.com/hsl/hschain/store/mq"
)

// 消息队列：多对多消息队列
// 消息：支持worker-task, topic模式

// 1. 队列特点：
// 1.1 每个topic 可以有多个订阅者；全局只有一个queue队列实例
// 1.2 消息的回复直接通过消息自带的channel回复

const (
	defaultHighChanBuffer = 64
	defaultLowChanBuffer  = 40960
	// DefaultQueueName default queue name
	DefaultQueueName = "default"
)

// 消息队列的错误
var (
	// ErrIsQueueClosed error queue is closed
	ErrIsQueueClosed = errors.New("ErrIsQueueClosed")
	// ErrQueueTimeout error queue timeout
	ErrQueueTimeout = errors.New("ErrQueueTimeout")
	// ErrQueueChannelFull error queue channel full
	ErrQueueChannelFull = errors.New("ErrQueueChannelFull")
)

type chanSub struct {
	high    chan *Message
	low     chan *Message
	isClose int32
}

// Queue only one obj in project
// Queue only generate Client and start、Close operate,
// if you send massage or receive massage on Queue, please use Client.
type Queue interface {
	Close()
	Start()
	Client(mode string) Client // 新建客户端实例
	Name() string
}

type queue struct {
	chanSubs map[string]*chanSub
	//chanWorkerSubs map[string]*chanSub
	//chanTopicSubs  map[string]*chanSub
	mu        sync.Mutex
	done      chan struct{}
	interrupt chan struct{}
	callback  chan *Message
	isClose   int32
	name      string
	msgPool   *sync.Pool
}

// NewQueue new queue struct
func NewQueue(name string) Queue {
	q := &queue{
		chanSubs: make(map[string]*chanSub),
		//chanTopicSubs:  make(map[string]*chanSub),
		//chanWorkerSubs: make(map[string]*chanSub),
		name:      name,
		done:      make(chan struct{}, 1),
		interrupt: make(chan struct{}, 1),
		callback:  make(chan *Message, 1024),
	}
	q.msgPool = &sync.Pool{
		New: func() interface{} {
			return &Message{
				chReply: make(chan *Message, 1),
			}
		},
	}
	go func() {
		for {
			select {
			case <-q.done:
				// log.Info("closing callback")
				return
			case msg := <-q.callback:
				if msg.callback != nil {
					msg.callback(msg)
				}
			}
		}
	}()
	return q
}

// Name return the queue name
func (q *queue) Name() string {
	return q.name
}

// Start 开始运行消息队列
func (q *queue) Start() {
	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
	// Block until a signal is received.
	select {
	case <-q.done:
		// log.Info("closing done")
		// atomic.StoreInt32(&q.isClose, 1)
		break
	case <-q.interrupt:
		// log.Info("closing")
		// atomic.StoreInt32(&q.isClose, 1)
		break
	case <-c:
		// log.Info("Got signal:", s)
		// atomic.StoreInt32(&q.isClose, 1)
		break
	}
}

func (q *queue) isClosed() bool {
	return atomic.LoadInt32(&q.isClose) == 1
}

// Close 关闭消息队列
func (q *queue) Close() {
	if q.isClosed() {
		return
	}
	q.mu.Lock()
	for topic, ch := range q.chanSubs {
		if ch.isClose == 0 {
			ch.high <- &Message{}
			ch.low <- &Message{}
			q.chanSubs[topic] = &chanSub{isClose: 1}
		}
	}
	//for topic, ch := range q.chanTopicSubs {
	//	if ch.isClose == 0 {
	//		ch.high <- &Message{}
	//		ch.low <- &Message{}
	//		q.chanTopicSubs[topic] = &chanSub{isClose: 1}
	//	}
	//}
	//
	//for worker, ch := range q.chanWorkerSubs {
	//	if ch.isClose == 0 {
	//		ch.high <- &Message{}
	//		ch.low <- &Message{}
	//		q.chanWorkerSubs[worker] = &chanSub{isClose: 1}
	//	}
	//}
	q.mu.Unlock()
	q.done <- struct{}{}
	close(q.done)
	atomic.StoreInt32(&q.isClose, 1)
	// log.Info("queue module closed")
}

// chanSub 获取chanSub（不存在则创建，用作支持先发送后订阅场景）
func (q *queue) chanSub(topic string) *chanSub {
	q.mu.Lock()
	defer q.mu.Unlock()

	if _, ok := q.chanSubs[topic]; !ok {
		q.chanSubs[topic] = &chanSub{
			high:    make(chan *Message, defaultHighChanBuffer),
			low:     make(chan *Message, defaultLowChanBuffer),
			isClose: 0,
		}
	}
	return q.chanSubs[topic]
}

// chanSub 获取chanSubTopic（不存在则创建，用作支持先发送后订阅场景）
//func (q *queue) chanSubTopic(topic string) *chanSub {
//	q.mu.Lock()
//	defer q.mu.Unlock()
//
//	if _, ok := q.chanTopicSubs[topic]; !ok {
//		q.chanTopicSubs[topic] = &chanSub{
//			high:    make(chan *Message, defaultHighChanBuffer),
//			low:     make(chan *Message, defaultLowChanBuffer),
//			isClose: 0,
//		}
//	}
//	return q.chanTopicSubs[topic]
//}

// 依据topic前缀随机获取一个chanSub
func (q *queue) chanSubTopicPrefixRandOne(topic string) *chanSub {
	q.mu.Lock()
	defer q.mu.Unlock()

	if _, ok := q.chanSubs[topic]; !ok {
		q.chanSubs[topic] = &chanSub{
			high:    make(chan *Message, defaultHighChanBuffer),
			low:     make(chan *Message, defaultLowChanBuffer),
			isClose: 0,
		}
	}

	return q.chanSubs[topic]
}

// 依据topic前缀获取chanSub数组
func (q *queue) chanSubTopicPrefix(topic string) []*chanSub {
	q.mu.Lock()
	defer q.mu.Unlock()

	var results []*chanSub
	for k, v := range q.chanSubs {
		if k != topic && strings.HasPrefix(k, topic) {
			results = append(results, v)
		}
	}
	return results
}

func (q *queue) closeTopic(topic string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	sub, ok := q.chanSubs[topic]
	if !ok {
		return
	}
	if sub.isClose == 0 {
		sub.high <- &Message{}
		sub.low <- &Message{}
	}
	q.chanSubs[topic] = &chanSub{isClose: 1}
	delete(q.chanSubs, topic)
}

//func (q *queue) closeTopicSub(topic string) {
//	q.mu.Lock()
//	defer q.mu.Unlock()
//	sub, ok := q.chanTopicSubs[topic]
//	if !ok {
//		return
//	}
//	if sub.isClose == 0 {
//		sub.high <- &Message{}
//		sub.low <- &Message{}
//	}
//	q.chanTopicSubs[topic] = &chanSub{isClose: 1}
//}
//
//func (q *queue) closeWorkerSub(topic string) {
//	q.mu.Lock()
//	defer q.mu.Unlock()
//	sub, ok := q.chanWorkerSubs[topic]
//	if !ok {
//		return
//	}
//	if sub.isClose == 0 {
//		sub.high <- &Message{}
//		sub.low <- &Message{}
//	}
//	q.chanWorkerSubs[topic] = &chanSub{isClose: 1}
//}

// send 发送消息 worker(task)模式-ModeWorker
func (q *queue) send(msg *Message, timeout time.Duration) (err error) {
	if q.isClosed() {
		return types.ErrChannelClosed
	}
	sub := q.chanSubTopicPrefixRandOne(msg.Topic)
	if sub.isClose == 1 {
		return types.ErrChannelClosed
	}
	if timeout == -1 {
		sub.high <- msg
		return nil
	}

	defer func() {
		res := recover()
		if res != nil {
			err = res.(error)
		}
	}()

	if timeout == 0 {
		select {
		case sub.high <- msg:
			return nil
		default:
			// log.Error("send chain full", "msg", msg, "topic", msg.Topic, "sub", sub)
			return ErrQueueChannelFull
		}
	}

	select {
	case sub.high <- msg:
	case <-time.After(timeout):
		// log.Error("send timeout", "msg", msg, "topic", msg.Topic, "sub", sub)
		return ErrQueueTimeout
	}
	return nil
}

func (q *queue) sendAsync(msg *Message) error {
	if q.isClosed() {
		return types.ErrChannelClosed
	}

	sub := q.chanSub(msg.Topic)
	if sub.isClose == 1 {
		return types.ErrChannelClosed
	}
	select {
	case sub.low <- msg:
		return nil
	default:
		// log.Error("send async err", "msg", msg, "err", ErrQueueChannelFull)
		return ErrQueueChannelFull
	}
}

// sendLowTimeout 发送低优先级消息 worker(task)模式-ModeWorker
func (q *queue) sendLowTimeout(msg *Message, timeout time.Duration) (err error) {
	if q.isClosed() {
		return types.ErrChannelClosed
	}

	sub := q.chanSubTopicPrefixRandOne(msg.Topic)
	if sub.isClose == 1 {
		return types.ErrChannelClosed
	}
	if timeout == -1 {
		sub.low <- msg
		return nil
	}
	if timeout == 0 {
		return q.sendAsync(msg)
	}

	select {
	case sub.low <- msg:
		return nil
	case <-time.After(timeout):
		// log.Error("send async timeout", "msg", msg)
		return ErrQueueTimeout
	}
}

// sendLowTimeoutWithTopicMode 发送低优先级消息 topic模式-ModeTopic
func (q *queue) sendLowTimeoutWithTopicMode(msg *Message, timeout time.Duration) (err error) {
	if q.isClosed() {
		return types.ErrChannelClosed
	}

	for _, sub := range q.chanSubTopicPrefix(msg.Topic) {
		if sub == nil {
			return types.ErrChannelClosed
		}
		if sub.isClose == 1 {
			err = types.ErrChannelClosed
			continue
		}
		// 如果缓存已经满了丢弃最前面的消息
		if len(sub.low) == defaultLowChanBuffer {
			<-sub.low
		}
		if timeout == -1 {
			sub.low <- msg
			continue
		}
		if timeout == 0 {
			err = q.sendAsync(msg)
			continue
		}

		select {
		case sub.low <- msg:
			continue
		case <-time.After(timeout):
			// log.Error("send async timeout", "msg", msg)
			err = ErrQueueTimeout
			continue
		}
	}

	return err
}

// Client 新建客户端实例
func (q *queue) Client(mode string) Client {
	return newClient(q, mode)
}

// Message message struct
type Message struct {
	Topic      string
	Ty         int64
	ID         int64
	Data       interface{}
	chReply    chan *Message
	chMsgReply chan *drivers.Message
	callback   func(msg *Message)
}

// NewMessage new message
func NewMessage(id int64, topic string, ty int64, data interface{}) (msg *Message) {
	msg = &Message{}
	msg.ID = id
	msg.Ty = ty
	msg.Data = data
	msg.Topic = topic
	msg.chReply = make(chan *Message, 1)
	return msg
}

// NewMessageCallback reply block
func NewMessageCallback(id int64, topic string, ty int64, data interface{}, callback func(msg *Message)) (msg *Message) {
	msg = &Message{}
	msg.ID = id
	msg.Ty = ty
	msg.Data = data
	msg.Topic = topic
	msg.callback = callback
	return msg
}

// GetData get message data
func (msg *Message) GetData() interface{} {
	if _, ok := msg.Data.(error); ok {
		return nil
	}
	return msg.Data
}

// Err if err return error msg, or return nil
func (msg *Message) Err() error {
	if err, ok := msg.Data.(error); ok {
		return err
	}
	return nil
}

// Reply reply message to reply chan
func (msg *Message) Reply(replyMsg *Message) {
	if msg.chMsgReply != nil {
		mainTopic, subTopic := splitTopicKey(replyMsg.Topic)
		c := &drivers.Message{MainTopic: mainTopic, SubTopic: subTopic, Ty: replyMsg.Ty, Data: replyMsg.Data, Err: replyMsg.Err()}
		msg.chMsgReply <- c
	}

	if msg.chReply == nil {
		// log.Debug("reply a empty ch reply", "msg", msg)
		return
	}
	msg.chReply <- replyMsg
}

// String print the message information
func (msg *Message) String() string {
	return fmt.Sprintf("{topic:%s, Ty:%s, Id:%d, Err:%v, Ch:%v}", msg.Topic,
		types.GetEventName(int(msg.Ty)), msg.ID, msg.Err(), msg.chReply != nil)
}

// ReplyErr reply error
func (msg *Message) ReplyErr(_ string, err error) {
	var reply prototypes.Reply
	if err != nil {
		// log.Error(title, "reply.err", err.Error())
		reply.IsOk = false
		reply.Msg = []byte(err.Error())
	} else {
		// log.Debug(title, "success", "ok")
		reply.IsOk = true
	}
	id := atomic.AddInt64(&gid, 1)
	msg.Reply(NewMessage(id, "", types.EventReply, &reply))
}

// SetMsgChan set msg to chan
func (msg *Message) SetMsgChan(c chan *drivers.Message) {
	msg.chMsgReply = c
}
